package com.ruyuan.engine.main;
import com.alibaba.fastjson.JSON;
import com.ruyuan.engine.pojo.ClientLog;
import com.ruyuan.engine.pojo.RuleResult;
import com.ruyuan.engine.pojo.advance.MultipleConditionPartitionRule;
import com.ruyuan.engine.pojo.advance.RuleBroadcastTableEnhance;
import com.ruyuan.engine.pojo.advance.RuleComplexBeanEnhance;
import com.ruyuan.engine.utils.Constants;
import com.ruyuan.engine.utils.FlinkKafkaUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.time.Duration;
import java.util.List;

/**
 * 处理动态规则的流程
 * */
public class RuleProcessEntryPoint {

    public static MapStateDescriptor<String, List<RuleComplexBeanEnhance> > mapRuleStateDesc = new MapStateDescriptor<>("rule_broadcast_state", TypeInformation.of(String.class),TypeInformation.of(new TypeHint<List<RuleComplexBeanEnhance>>() {}));

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        /**
         * 用户数据：CLIENT_LOG
         * */
        DataStreamSource<String> source = env.addSource(FlinkKafkaUtils.getKafkaEventSource(Constants.CLIENT_LOG));
        //转换Json和过滤
        SingleOutputStreamOperator<ClientLog> clientSource = source.map(m -> JSON.parseObject(m, ClientLog.class)).filter(e -> e != null);
        //设置watermark
        WatermarkStrategy<ClientLog> clientLogWatermarkStrategy = WatermarkStrategy.<ClientLog>forBoundedOutOfOrderness(Duration.ofMillis(0)).withTimestampAssigner(new SerializableTimestampAssigner<ClientLog>() {
            @Override
            public long extractTimestamp(ClientLog clientLog, long recordTimestamp) {
                return clientLog.getCreateTime();
            }
        });
        clientSource.assignTimestampsAndWatermarks(clientLogWatermarkStrategy);

        /**
         * 规则流
         * */
        DataStreamSource<String> ruleSource = env.addSource(FlinkKafkaUtils.getKafkaEventSource(Constants.RULE_DROOLS));
        SingleOutputStreamOperator<RuleBroadcastTableEnhance> ruleBroadcastTableSingleOutputStreamOperator = ruleSource.map(o -> JSON.parseObject(o, RuleBroadcastTableEnhance.class));
        //规则广播
        BroadcastStream<RuleBroadcastTableEnhance> broadcastTableBroadcastStream = ruleBroadcastTableSingleOutputStreamOperator.broadcast(mapRuleStateDesc);

        //将业务数据和规则进行conncet
        BroadcastConnectedStream<ClientLog, RuleBroadcastTableEnhance> broadcastTableBroadcastConnectedStream = clientSource.connect(broadcastTableBroadcastStream);

        SingleOutputStreamOperator<MultipleConditionPartitionRule> multipleConditionPartitionRuleSingleOutputStreamOperator = broadcastTableBroadcastConnectedStream.process(new DynamicPartitionByRuleCondition());

        /**
         * 动态分区，进行规则匹配
         * */
        KeyedStream<MultipleConditionPartitionRule, String> multipleConditionPartitionRuleStringKeyedStream = multipleConditionPartitionRuleSingleOutputStreamOperator.keyBy(m -> m.getPartitionValues());

        /**
         * 规则计算
         * */
        BroadcastConnectedStream<MultipleConditionPartitionRule, RuleBroadcastTableEnhance> connectResult = multipleConditionPartitionRuleStringKeyedStream.connect(broadcastTableBroadcastStream);
        SingleOutputStreamOperator<RuleResult> process = connectResult.process(new RuleProcessFunction());
        /**
         * 结果可以输出到惩罚系统对应的Topic
         * */
        process.print();
        //最终计算
        env.execute();
    }
}